package belf.migrate.api.source.cdc;

import belf.migrate.api.exception.DDLEventNotSupport;
import belf.migrate.api.exception.InvalidChangedEvent;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.extern.slf4j.Slf4j;

/**
 * 将debezium发布的ChangedEvent事件转换为精简的json格式，方便存放到队列
 */
@Slf4j
public class CDCEventConverter {

    /**
     * 将收到的debezium的ChangedEvent事件转换为精炼的json对象，存放到redis队列。
     * @param key
     * @param value
     * @return
     */
    public static String to(String key, String value) {
        ObjectMapper mapper = new ObjectMapper();
        try {
            ObjectNode mergedNode = (ObjectNode)mapper.readTree("{}");

            if (key == null) {
                mergedNode.set("key", null);
            } else {
                JsonNode keyNode = mapper.readTree(key);
                mergedNode.set("key", keyNode);
            }
            ObjectNode valueNode = (ObjectNode)mapper.readTree(value);
            if (valueNode.get("ddl") != null) {
                throw new DDLEventNotSupport(valueNode.get("ddl").toString());
            }
            if (valueNode.get("op") == null) { //op: c,u,d,r
                throw new InvalidChangedEvent(value);
            }

            mergedNode.set("before", valueNode.get("before"));
            mergedNode.set("after", valueNode.get("after"));
            mergedNode.set("op", valueNode.get("op"));
            mergedNode.set("db", valueNode.get("source").get("db"));
            mergedNode.set("table", valueNode.get("source").get("table"));
            return mergedNode.toPrettyString();

        } catch (Exception e) {
//            e.printStackTrace();
//            log.error("Parse ChangedEvent error: {}, stacktrace:\n{}", e.getMessage(), e);
        }

        return null;
    }


}
